// This file is part of OpenTSDB.
// Copyright (C) 2017 TeleDB Development Group.
//
// This program is free software: you can redistribute it and/or modify it
// under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 2.1 of the License, or (at your
// option) any later version.  This program is distributed in the hope that it
// will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty
// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser
// General Public License for more details.  You should have received a copy
// of the GNU Lesser General Public License along with this program.  If not,
// see <http://www.gnu.org/licenses/>.
package net.opentsdb.tsd.points;

import com.stumbleupon.async.Deferred;
import net.opentsdb.core.Const;
import net.opentsdb.core.IncomingDataPoint;
import net.opentsdb.core.TSDB;
import net.opentsdb.core.Tags;
import org.hbase.async.Bytes;

/**
 * 带配置写入，只支持put
 *
 */
public class PutPointWithConf {

    static DataPointType getType(IncomingDataPoint dp){
            return DataPointType.PUT;
    }

    public static Deferred<Object> addPoint(TSDB tsdb, IncomingDataPoint dp, PutConfig config){
        AddPointFactory factory = AddPointFactory.getAddPoint(dp);
        AddPoint addPoint = factory.createAddPoint(tsdb ,dp, config);
        return addPoint.addPoint();
    }

    enum AddPointFactory{
        PutInteger{AddPoint createAddPoint(TSDB tsdb, IncomingDataPoint dp, PutConfig config) {return new PutIntegerAddPoint(tsdb, dp, config);}},
        PutDouble{AddPoint createAddPoint(TSDB tsdb, IncomingDataPoint dp, PutConfig config) {return new PutDoubleAddPoint(tsdb, dp, config);}};

       static AddPointFactory getAddPoint(IncomingDataPoint dp){
           boolean integerFlag = Tags.looksLikeInteger(dp.getValue());

           switch (getType(dp)){
               case PUT:
                   if(integerFlag) return PutInteger;
                   else return PutDouble;
               default:
                   return null;
           }
       }

       abstract AddPoint createAddPoint(TSDB tsdb, IncomingDataPoint dp, PutConfig config);
    }

    interface AddPoint{
        Deferred<Object> addPoint();
    }

    abstract static class AbstractAddPoint implements AddPoint {
        final TSDB tsdb;
        final IncomingDataPoint dp;
        final long timestamp;
        final PutConfig config;

        public AbstractAddPoint(TSDB tsdb, IncomingDataPoint dp, PutConfig config){
            this.tsdb = tsdb;
            this.dp = dp;
            this.config = config;
            long time;
            // 时间统一处理为13位
            time = (dp.getTimestamp() & Const.SECOND_MASK)!=0?dp.getTimestamp():dp.getTimestamp()*1000;

            //按照period时间规整
            if(tsdb.getConfig().isEnableRegularTime()){
                long period = tsdb.getConfig().getRegularTimePeriod();
                if(period > 0){
                    time = time - time%(period*1000);
                }
            }
            timestamp = time;
        }
    }

    static class PutIntegerAddPoint extends AbstractAddPoint {

        public PutIntegerAddPoint(TSDB tsdb, IncomingDataPoint dp, PutConfig config) {
            super(tsdb, dp, config);
        }

        @Override
        public Deferred<Object> addPoint() {

            long value = Tags.parseLong(dp.getValue());

            final byte[] v;
            if (Byte.MIN_VALUE <= value && value <= Byte.MAX_VALUE) {
                v = new byte[] { (byte) value };
            } else if (Short.MIN_VALUE <= value && value <= Short.MAX_VALUE) {
                v = Bytes.fromShort((short) value);
            } else if (Integer.MIN_VALUE <= value && value <= Integer.MAX_VALUE) {
                v = Bytes.fromInt((int) value);
            } else {
                v = Bytes.fromLong(value);
            }

            final short flags = (short) (v.length - 1);  // Just the length.

            PutDataContext context = PutDataContext
                    .newBuilder(dp.getMetric(),dp.getTags(),timestamp,v)
                    .flags(flags)
                    .forward(config.getForward())
                    .build();

            return tsdb.addPointInternal(context);

        }
    }

    static class PutDoubleAddPoint extends AbstractAddPoint {

        public PutDoubleAddPoint(TSDB tsdb, IncomingDataPoint dp, PutConfig config) {
            super(tsdb, dp, config);
        }

        @Override
        public Deferred<Object> addPoint() {
            Double value = (Tags.fitsInFloat(dp.getValue()) ?
                    Float.parseFloat(dp.getValue()) :
                    Double.parseDouble(dp.getValue()));

            if (Double.isNaN(value) || Double.isInfinite(value)) {
                throw new IllegalArgumentException("value is NaN or Infinite: " + value
                        + " for metric=" + dp.getMetric()
                        + " timestamp=" + timestamp);
            }
            final short flags = Const.FLAG_FLOAT | 0x7;  // A float stored on 8 bytes.

            PutDataContext context = PutDataContext
                    .newBuilder(dp.getMetric(),
                            dp.getTags(),
                            timestamp,
                            Bytes.fromLong(Double.doubleToRawLongBits(value)))
                    .flags(flags)
                    .forward(config.getForward())
                    .build();

            return tsdb.addPointInternal(context);
        }
    }



    public enum DataPointType {
        PUT
    }
}
